package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.udf.ArrayDeSer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Etl08_AdviceShowClickFeatureSession {

    public static void main(String[] args) {

        // 编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka连接器表，来映射 kafka 中的dwd层的用户行为公共维度宽表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                           "
                        + "     user_id           BIGINT,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     properties        map<string,string>,         "
                        + "     pt AS proctime() ,                            "
                        + "     rt AS to_timestamp_ltz(event_time,3),         "
                        + "     WATERMARK FOR rt AS rt - interval '0' second  "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");

        /**
         * 表结构扁平化
         */
        tenv.executeSql(
                " CREATE TEMPORARY VIEW tmp_view AS                  "
                        +" SELECT                                             "
                        +"   user_id,                                         "
                        +"   event_id,                                        "
                        +"   event_time,                                      "
                        +"   properties['creative_id'] as ad_creative_id,     "
                        +"   properties['ad_tracking_id'] as ad_tracking_id,  "
                        +"   rt,                                              "
                        +"   pt                                               "
                        +" FROM dwd_kafka                                     "
                        +" WHERE event_id in ('ad_show','ad_click')           "
        );


        /**
         * 流内模式匹配
         */
        tenv.executeSql(
                " CREATE TEMPORARY VIEW show_click  AS                       "
                +" SELECT                                                     "
                        +"   uid as user_id,                                 "
                        +"   tk_id,                                          "
                        +"   ad_id,                                          "
                        +"   show_time,                                      "
                        +"   click_time                                      "
                        +" FROM tmp_view                                     "
                        +"   MATCH_RECOGNIZE(                                "
                        +" 	PARTITION BY user_id                             "
                        +" 	ORDER BY pt                                      "
                        +" 	MEASURES                                         "
                        +" 	  A.user_id as uid,                              "
                        +"       A.ad_creative_id as ad_id,                  "
                        +"       A.event_time as show_time,                  "
                        +"       A.ad_tracking_id as tk_id,                  "
                        +"       B.event_time as click_time	  	             "
                        +" 	ONE ROW PER MATCH                                "
                        +"     AFTER MATCH SKIP TO NEXT ROW  	             "
                        +"     PATTERN (A M*? B) WITHIN INTERVAL '5' SECOND  "
                        +" 	DEFINE                                           "
                        +" 	  A AS A.event_id = 'ad_show',                   "
                        +" 	  M AS M.ad_tracking_id IS NOT NULL,             "
                        +" 	  B AS B.event_id = 'ad_click' AND B.ad_tracking_id =  A.ad_tracking_id  "
                        +" 	  )                                              "
        );

        tenv.executeSql("select * from show_click").print();

        /*tenv.executeSql(
                " CREATE TEMPORARY VIEW show_click  AS  SELECT              "
                        +"   uid as user_id,                                 "
                        +"   tk_id,                                          "
                        +"   ad_id,                                          "
                        +"   show_time,                                      "
                        +"   click_time,                                      "
                        +"   apt as pt                                        "
                        +" FROM tmp_view                                     "
                        +"   MATCH_RECOGNIZE(                                "
                        +" 	PARTITION BY ad_tracking_id                      "
                        +" 	ORDER BY pt                                      "
                        +" 	MEASURES                                         "
                        +" 	  A.user_id as uid,                              "
                        +"    A.ad_creative_id as ad_id,                     "
                        +"    A.event_time as show_time,                     "
                        +"    A.ad_tracking_id as tk_id,                     "
                        +"    A.pt as apt,                                   "
                        +"    B.event_time as click_time	  	             "
                        +" 	ONE ROW PER MATCH                                "
                        +"     AFTER MATCH SKIP PAST LAST ROW	             "
                        +"     PATTERN (A B) WITHIN INTERVAL '10' MINUTE     "
                        +" 	DEFINE                                           "
                        +" 	  A AS A.event_id = 'ad_show',                   "
                        +" 	  B AS B.event_id = 'ad_click'                   "
                        +" 	  )                                              "
        );*/







        // TODO   关联 请求特征流数据
        // 创建hbase连接器表，映射hbase中的请求特征日志数据表
        tenv.executeSql(
                "CREATE TABLE request_hbase ( " +
                        " ad_tracking_id   STRING, " +
                        " f ROW<ad_id STRING,ad_features BYTES,user_features BYTES,requestTime BIGINT>, " +
                        " PRIMARY KEY (ad_tracking_id) NOT ENFORCED " +
                        ") WITH (                             " +
                        " 'connector' = 'hbase-2.2',          " +
                        " 'table-name' = 'ad_request_log',     " +
                        " 'zookeeper.quorum' = 'doitedu:2181' " +
                        ")");

        // 创建kafka连接器表，用来映射接收最终结果数据的topic
        tenv.executeSql(
                "  CREATE TABLE ad_feature_session (                           "
                        + "     user_id           BIGINT,                      "
                        + "     tk_id             STRING,                     "
                        + "     ad_id             STRING,                     "
                        + "     show_time         BIGINT,                     "
                        + "     click_time        BIGINT,                     "
                        + "     requestTime       BIGINT,                     "
                        + "     ad_features       ARRAY<DOUBLE>,              "
                        + "     user_features     ARRAY<DOUBLE>               "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'ad-feature-session',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");


        tenv.createTemporaryFunction("des", ArrayDeSer.class);
        tenv.executeSql(
                " INSERT INTO ad_feature_session                                      "
                        +" SELECT  /*+ LOOKUP('table'='request_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='1s','max-attempts'='3') */   "
                        +" 	sc.user_id,                                               "
                        +" 	sc.tk_id,                                                 "
                        +" 	sc.ad_id,                                                 "
                        +" 	sc.show_time,                                             "
                        +" 	sc.click_time,                                            "
                        +" 	re.requestTime,                                           "
                        +" 	des(re.ad_features) as ad_features,                       "
                        +" 	des(re.user_features) as user_features                    "
                        +" FROM show_click sc                                         "
                        +" LEFT JOIN request_hbase FOR SYSTEM_TIME AS OF sc.pt AS re  "
                        +" ON sc.tk_id = re.ad_tracking_id                            "
        );

    }
}
